package com.yeeiee.factory;

import lombok.val;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.catalog.exceptions.CatalogException;

import java.util.Set;
import java.util.stream.Collectors;

/**
 * 工厂
 *
 * @param <T> 生产对象
 */
public interface ClinkFactory<T> {

    /**
     * 标识符
     */
    String factoryIdentifier();

    /**
     * 必要的配置
     */
    Set<ConfigOption<?>> requiredOptions();

    /**
     * 创建对象
     *
     * @param configuration 配置
     */
    T build(Configuration configuration);

    /**
     * 校验配置
     *
     * @param configuration 配置
     */
    default void validate(Configuration configuration) {
        val missingOptions = requiredOptions().stream().filter(option -> !configuration.contains(option)).map(ConfigOption::key).collect(Collectors.toList());
        if (!missingOptions.isEmpty()) {
            throw new CatalogException("Missing required options: " + missingOptions);
        }
    }
}
